package com.example.tms.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.tms.realtime.app.fun.MyBroadcastProcessFunction;
import com.example.tms.realtime.beans.TmsConfigDimBean;
import com.example.tms.realtime.common.TmsConfig;
import com.example.tms.realtime.utils.CreateEnvUtil;
import com.example.tms.realtime.utils.HbaseUtil;
import com.example.tms.realtime.utils.KafKaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.codehaus.plexus.util.StringUtils;

/**
 * Created with IntelliJ IDEA.
 * ClassName: DimApp
 * Package: com.example.tms.realtime.app.dim
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-01
 * Time: 17:40
 */

public class DimApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境以及检查点相关配置
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(args);
        //1.2 设置并行度
        env.setParallelism(4);
        //kafka的分区数 和 这个并行度保持一致
        //TODO 2.从kafka的tms_ods主题中读取业务数据
        //2.1 现在是消费数据 声明消费者组
        String topic = "tms_ods";
        String groupId = "dim_app_group";
        //2.2 创建消费者对象 要维护一个偏移量
        KafkaSource<String> kafkaSource = KafKaUtil.getKafkaSource(topic, groupId, args);
        //2.3 消费数据 封装为流
        SingleOutputStreamOperator<String> kafkaStrDS = env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka_source")
                .uid("kafka_source"); //有状态ID指定uid

        //验证能不能读取成功 要先开启消费者 在开生产者
        //kafkaStrDS.print(">>>");
        //TODO 3.对读取的数据进行类型转换并过滤掉不需要传递的json属性
        //向下游传递数据通过拿出的数据分析 有些数据是无用的 就可以不用下下游传递
        final SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(new MapFunction<String, JSONObject>() {
            //map算子 灵活运用也可以用来过滤
            @Override
            public JSONObject map(String value) throws Exception {
                //为了操作对象 又json对象字符串转换为json对象
                JSONObject jsonObject = JSON.parseObject(value);
                //在source对象中只有table属性有用 获取到table值
                String table = jsonObject.getJSONObject("source").getString("table");
                //删除json中的一些数据
                jsonObject.remove("before");
                jsonObject.remove("source");
                jsonObject.remove("transaction");
                //在将table放入到数据中
                jsonObject.put("table", table);
                return jsonObject;

            }
        });
        //jsonObjDS.print("<<<");

        //TODO 4.使用FlinkCDC读取配置表数据
        //4.1 获取MysqlSource数据源 ??? 为什么还是数据库
        MySqlSource<String> mySqlSource = CreateEnvUtil.getMySqlSource("config_dim","6000",args);
        //4.2 获取数据 封装为流
        SingleOutputStreamOperator<String> mysqlDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source")
                //并行度 设置为 1 上面的serverID只有一个 但是全局并行度是4 报错
                .setParallelism(1)
                .uid("mysql_source");
        //现在已经是两条流了 配置表的数据 一个是Kafka的数据
        mysqlDS.print();

        //TODO 5.提前将hbase中的维度表创建出来
        final SingleOutputStreamOperator<String> createTableDS = mysqlDS.map(
                new MapFunction<String, String>() {
                    @Override
                    public String map(String jsonStr) throws Exception {
                        //将JSON字符串 转换成JSON对象
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        //获取配置表的操作
                        String op = jsonObject.getString("op");
                        if ("r".equals(op) || "c".equals(op)) {
                            //获取表明 在JSON对象当中
                            JSONObject after = jsonObject.getJSONObject("after");
                            String sinkTable = after.getString("sink_table");
                            //获取列族
                            String sinkFamily = after.getString("sink_family");
                            if (StringUtils.isEmpty(sinkFamily)) {
                                sinkFamily = "info";
                            }

                            System.out.println("在Hbase中创建表" + sinkTable);
                            //第一个参数是库名
                            HbaseUtil.createTable(TmsConfig.HBASE_NAMESPACE, sinkTable, sinkFamily.split(","));
                        }

                        return jsonStr;
                    }
                }
        );
        //createTableDS.print("<<<");
        //TODO 6.对配置数据进行广播
        MapStateDescriptor<String, TmsConfigDimBean> mapStateDes = new MapStateDescriptor<String, TmsConfigDimBean>("mapStateDescriptor",String.class,TmsConfigDimBean.class);
        BroadcastStream<String> broadcastDS = createTableDS.broadcast(mapStateDes);
        //TODO 7.将主流和广播进行关联 --connect算子
        BroadcastConnectedStream<JSONObject, String> connect = jsonObjDS.connect(broadcastDS);
        SingleOutputStreamOperator<JSONObject> dimDS = connect.process(new MyBroadcastProcessFunction(mapStateDes));
        //TODO 8.对关联之后的数据进行处理
        dimDS.print("<<<");
        //TODO 9.将维度数据保存到hbase表中

        env.execute();
    }
}